package org.jetlinks.community.messaging.kafka.rule;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import org.hswebframework.web.bean.FastBeanCopier;
import org.hswebframework.web.validator.ValidatorUtils;
import org.jetlinks.community.datasource.DataSourceManager;
import org.jetlinks.community.messaging.kafka.Message;
import org.jetlinks.community.messaging.kafka.impl.KafkaDataSourceProvider;
//import org.jetlinks.community.rule.engine.editor.annotation.EditorResource;
//import org.jetlinks.community.rule.engine.model.nodes.NodeConverter;
import org.jetlinks.rule.engine.api.RuleConstants;
import org.jetlinks.rule.engine.api.RuleData;
import org.jetlinks.rule.engine.api.model.RuleNodeModel;
import org.jetlinks.rule.engine.api.task.ExecutionContext;
import org.jetlinks.rule.engine.api.task.TaskExecutor;
import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
import org.jetlinks.rule.engine.defaults.AbstractTaskExecutor;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.stereotype.Component;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;

import javax.validation.constraints.NotBlank;
import java.nio.charset.StandardCharsets;
import java.util.*;

@Component
@AllArgsConstructor
//@EditorResource(
//    id = "kafka-consumer",
//    name = "订阅Kafka",
//    editor = "rule-engine/editor/network/kafka-consumer.html",
//    helper = "rule-engine/i18n/{language}/network/kafka-consumer.html",
//    order = 200
//)
//public class KafkaConsumerTaskExecutorProvider implements TaskExecutorProvider, NodeConverter {
public class KafkaConsumerTaskExecutorProvider implements TaskExecutorProvider {


    @Override
    public String getExecutor() {
        return "kafka-consumer";
    }

    @Override
    public Mono<TaskExecutor> createTask(ExecutionContext context) {
        return Mono.just(new KafkaConsumerTaskExecutor(context));
    }

    public String getNodeType() {
        return getExecutor();
    }

    public RuleNodeModel convert(JSONObject nodeJson) {
        RuleNodeModel model = new RuleNodeModel();
        model.setExecutor(getExecutor());
        model.setConfiguration(nodeJson);
        return model;
    }


    static class KafkaConsumerTaskExecutor extends AbstractTaskExecutor {

        private KafkaConsumerConfig config;

        public KafkaConsumerTaskExecutor(ExecutionContext context) {
            super(context);
            this.config = createConfig();
        }

        @Override
        public String getName() {
            return "Kafka Consumer";
        }

        @Override
        protected Disposable doStart() {
            return KafkaDataSourceProvider
                .create("rule", config.createProperties(String.join(":", "rule-engine", context
                    .getJob()
                    .getInstanceId(), context.getJob().getNodeId())))
                .createConsumer(config.parseTopic())
                .flatMapMany(consumer -> consumer
                    .subscribe()
                    .flatMap(this::handleMessage)
                    .doOnCancel(consumer::shutdown))
                .subscribe();
        }

        private Mono<Void> handleMessage(Message message) {
            try {
                String body = message.getPayload().toString(StandardCharsets.UTF_8);
                Object data;
                //json
                if (body.startsWith("{") || body.startsWith("[")) {
                    data = JSON.parse(body);
                } else {
                    Map<String, Object> map = new HashMap<>();
                    map.put("topic", message.getTopic());
                    map.put("payload", body);
                    data = map;
                }
                RuleData ruleData = context.newRuleData(data);

                return context
                    .getOutput()
                    .write(ruleData)
                    .then(context.fireEvent(RuleConstants.Event.result, ruleData))
                    .onErrorResume(err -> context.onError(err, ruleData))
                    .then();
            } catch (Throwable e) {
                return context.onError(e, null);
            }
        }

        public KafkaConsumerConfig createConfig() {
            return ValidatorUtils.tryValidate(
                FastBeanCopier
                    .copy(context.getJob().getConfiguration(), new KafkaConsumerConfig())
            );
        }

        @Override
        public void reload() {
            config = createConfig();
            if (disposable != null) {
                disposable.dispose();
            }
            this.disposable = doStart();
        }
    }

    @Getter
    @Setter
    public static class KafkaConsumerConfig {
        @NotBlank
        private String bootstrapServers;

        @NotBlank
        private String topic;

        @NotBlank
        private String groupId;

        List<String> parseTopic() {
            return Collections.singletonList(topic);
        }

        KafkaProperties createProperties(String clientId) {
            KafkaProperties source = new KafkaProperties();
            source.setBootstrapServers(Arrays.asList(bootstrapServers.split(",")));
            source.getConsumer().setGroupId(groupId);
            source.getConsumer().setClientId(clientId);
            return source;
        }
    }
}
